{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# 量化回测框架\n",
    "\n",
    "本笔记本实现统一的量化策略回测框架"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import pandas as pd\n",
    "import numpy as np\n",
    "import matplotlib.pyplot as plt\n",
    "from datetime import datetime\n",
    "\n",
    "# 设置中文显示\n",
    "plt.rcParams['font.sans-serif'] = ['SimHei']\n",
    "plt.rcParams['axes.unicode_minus'] = False"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 回测引擎核心类"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "class BacktestEngine:\n",
    "    \"\"\"\n",
    "    量化回测引擎\n",
    "    \n",
    "    功能：\n",
    "    1. 加载数据\n",
    "    2. 执行策略\n",
    "    3. 计算绩效\n",
    "    4. 生成报告\n",
    "    \"\"\"\n",
    "    def __init__(self, data, initial_capital=1000000, commission=0.0005):\n",
    "        \"\"\"\n",
    "        初始化回测引擎\n",
    "        \n",
    "        Args:\n",
    "            data: 包含价格数据的DataFrame\n",
    "            initial_capital: 初始资金\n",
    "            commission: 交易佣金率\n",
    "        \"\"\"\n",
    "        self.data = data\n",
    "        self.initial_capital = initial_capital\n",
    "        self.commission = commission\n",
    "        self.positions = 0\n",
    "        self.cash = initial_capital\n",
    "        self.equity = []\n",
    "        self.trades = []\n",
    "        \n",
    "    def run_backtest(self, strategy):\n",
    "        \"\"\"\n",
    "        运行回测\n",
    "        \n",
    "        Args:\n",
    "            strategy: 策略函数，需返回交易信号\n",
    "        \"\"\"\n",
    "        signals = strategy(self.data)\n",
    "        \n",
    "        for i in range(len(self.data)):\n",
    "            price = self.data.iloc[i]['close']\n",
    "            signal = signals.iloc[i] if i < len(signals) else 0\n",
    "            \n",
    "            # 执行交易\n",
    "            if signal != 0:\n",
    "                self._execute_trade(signal, price)\n",
    "                \n",
    "            # 更新权益\n",
    "            self._update_equity(price)\n",
    "            \n",
    "        return self._generate_report()\n",
    "    \n",
    "    def _execute_trade(self, signal, price):\n",
    "        \"\"\"执行交易\"\"\"\n",
    "        # 计算可交易数量\n",
    "        size = int(self.cash * 0.95 / price) if signal > 0 else self.positions\n",
    "        \n",
    "        if signal > 0:  # 买入\n",
    "            cost = size * price * (1 + self.commission)\n",
    "            if cost > self.cash:\n",
    "                return\n",
    "                \n",
    "            self.positions += size\n",
    "            self.cash -= cost\n",
    "            self.trades.append({\n",
    "                'date': self.data.index[i],\n",
    "                'type': 'buy',\n",
    "                'price': price,\n",
    "                'size': size\n",
    "            })\n",
    "        else:  # 卖出\n",
    "            proceeds = size * price * (1 - self.commission)\n",
    "            self.positions -= size\n",
    "            self.cash += proceeds\n",
    "            self.trades.append({\n",
    "                'date': self.data.index[i],\n",
    "                'type': 'sell',\n",
    "                'price': price,\n",
    "                'size': size\n",
    "            })\n",
    "    \n",
    "    def _update_equity(self, price):\n",
    "        \"\"\"更新权益曲线\"\"\"\n",
    "        self.equity.append(self.cash + self.positions * price)\n",
    "    \n",
    "    def _generate_report(self):\n",
    "        \"\"\"生成回测报告\"\"\"\n",
    "        returns = pd.Series(self.equity).pct_change().dropna()\n",
    "        \n",
    "        report = {\n",
    "            '初始资金': self.initial_capital,\n",
    "            '最终权益': self.equity[-1],\n",
    "            '总收益率': (self.equity[-1] - self.initial_capital) / self.initial_capital,\n",
    "            '年化收益率': self._annualized_return(returns),\n",
    "            '最大回撤': self._max_drawdown(),\n",
    "            '夏普比率': self._sharpe_ratio(returns),\n",
    "            '交易次数': len(self.trades)\n",
    "        }\n",
    "        \n",
    "        return report\n",
    "    \n",
    "    def _annualized_return(self, returns):\n",
    "        \"\"\"计算年化收益率\"\"\"\n",
    "        days = (self.data.index[-1] - self.data.index[0]).days\n",
    "        return (1 + returns.mean()) ** (365 / days) - 1\n",
    "    \n",
    "    def _max_drawdown(self):\n",
    "        \"\"\"计算最大回撤\"\"\"\n",
    "        equity = pd.Series(self.equity)\n",
    "        peak = equity.cummax()\n",
    "        drawdown = (equity - peak) / peak\n",
    "        return drawdown.min()\n",
    "    \n",
    "    def _sharpe_ratio(self, returns, rf=0.03):\n",
    "        \"\"\"计算夏普比率\"\"\"\n",
    "        excess_returns = returns - rf / 252\n",
    "        return excess_returns.mean() / excess_returns.std() * np.sqrt(252)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 示例策略"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def moving_average_crossover(data, short_window=10, long_window=50):\n",
    "    \"\"\"\n",
    "    均线交叉策略\n",
    "    \n",
    "    Args:\n",
    "        data: 价格数据\n",
    "        short_window: 短期均线窗口\n",
    "        long_window: 长期均线窗口\n",
    "        \n",
    "    Returns:\n",
    "        pd.Series: 交易信号(1:买入, -1:卖出, 0:无操作)\n",
    "    \"\"\"\n",
    "    signals = pd.DataFrame(index=data.index)\n",
    "    signals['price'] = data['close']\n",
    "    signals['short_ma'] = data['close'].rolling(short_window).mean()\n",
    "    signals['long_ma'] = data['close'].rolling(long_window).mean()\n",
    "    \n",
    "    # 生成信号\n",
    "    signals['signal'] = 0\n",
    "    signals['signal'][short_window:] = np.where(\n",
    "        signals['short_ma'][short_window:] > signals['long_ma'][short_window:], 1, 0)\n",
    "    \n",
    "    # 获取交易信号\n",
    "    signals['positions'] = signals['signal'].diff()\n",
    "    \n",
    "    return signals['positions']"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 主分析流程"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# 示例：加载数据\n",
    "# 这里使用模拟数据，实际应用中应从数据源获取\n",
    "dates = pd.date_range('2020-01-01', '2023-12-31')\n",
    "prices = np.cumprod(1 + np.random.normal(0.0005, 0.01, len(dates)))\n",
    "data = pd.DataFrame({'close': prices}, index=dates)\n",
    "\n",
    "# 初始化回测引擎\n",
    "engine = BacktestEngine(data)\n",
    "\n",
    "# 运行回测\n",
    "report = engine.run_backtest(lambda x: moving_average_crossover(x))\n",
    "\n",
    "# 输出结果\n",
    "print(\"回测结果:\")\n",
    "for k, v in report.items():\n",
    "    print(f\"{k}: {v:.4f}\" if isinstance(v, float) else f\"{k}: {v}\")\n",
    "\n",
    "# 绘制权益曲线\n",
    "plt.figure(figsize=(12, 6))\n",
    "plt.plot(engine.equity)\n",
    "plt.title('策略权益曲线')\n",
    "plt.xlabel('日期')\n",
    "plt.ylabel('权益')\n",
    "plt.grid()\n",
    "plt.show()"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.8.5"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 4
}